package com.example.flink;

import com.example.entity.FinancialIndicator;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;

/**
 * 财务指标Flink任务 - 暂时禁用
 */
@Slf4j
// @Component
public class FinancialIndicatorJob {

    @Autowired
    private StreamExecutionEnvironment env;

    @Value("${flink.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${flink.kafka.group-id}")
    private String groupId;

    @Value("${flink.kafka.topics.financial-data}")
    private String financialDataTopic;

    /**
     * 静态内部类 - 财务指标数据映射器
     */
    public static class FinancialIndicatorMapper implements MapFunction<String, FinancialIndicator> {
        private static final ObjectMapper objectMapper = new ObjectMapper();

        @Override
        public FinancialIndicator map(String value) throws Exception {
            try {
                JsonNode jsonNode = objectMapper.readTree(value);
                FinancialIndicator indicator = new FinancialIndicator();

                // 设置基本字段
                indicator.setUserId(jsonNode.path("userId").asLong());
                indicator.setMonthlyIncome(new BigDecimal(jsonNode.path("monthlyIncome").asText("0")));
                indicator.setIncomeType(jsonNode.path("incomeType").asInt(1));
                indicator.setHasHouse(jsonNode.path("hasHouse").asInt(0));
                indicator.setHouseValue(new BigDecimal(jsonNode.path("houseValue").asText("0")));
                indicator.setHasCar(jsonNode.path("hasCar").asInt(0));
                indicator.setCarValue(new BigDecimal(jsonNode.path("carValue").asText("0")));
                indicator.setDepositAmount(new BigDecimal(jsonNode.path("depositAmount").asText("0")));
                indicator.setDebtAmount(new BigDecimal(jsonNode.path("debtAmount").asText("0")));
                indicator.setCreateTime(LocalDateTime.now());
                indicator.setUpdateTime(LocalDateTime.now());

                return indicator;
            } catch (Exception e) {
                log.error("解析财务数据失败: {}", value, e);
                return null;
            }
        }
    }

    /**
     * 静态内部类 - 财务指标处理器
     */
    public static class FinancialIndicatorProcessor extends ProcessWindowFunction<FinancialIndicator, FinancialIndicator, Long, TimeWindow> {
        @Override
        public void process(Long userId, ProcessWindowFunction<FinancialIndicator, FinancialIndicator, Long, TimeWindow>.Context context,
                           Iterable<FinancialIndicator> elements, Collector<FinancialIndicator> out) {
            List<FinancialIndicator> indicators = new ArrayList<>();
            elements.forEach(indicators::add);

            if (indicators.isEmpty()) {
                return;
            }

            // 获取最新的财务指标数据
            FinancialIndicator latestIndicator = indicators.get(indicators.size() - 1);

            // 计算负债收入比（DTI）
            if (latestIndicator.getMonthlyIncome().compareTo(BigDecimal.ZERO) > 0) {
                BigDecimal dti = latestIndicator.getDebtAmount()
                        .divide(latestIndicator.getMonthlyIncome(), 4, RoundingMode.HALF_UP);
                latestIndicator.setDebtToIncomeRatio(dti);
            }

            // 计算可支配收入
            BigDecimal disposableIncome = latestIndicator.getMonthlyIncome()
                    .subtract(latestIndicator.getDebtAmount());
            latestIndicator.setDisposableIncome(disposableIncome);

            // 输出处理后的财务指标
            out.collect(latestIndicator);

            log.info("处理用户[{}]的财务指标数据完成", userId);
        }
    }

    public void init() {
        try {
            // 创建Kafka数据源
            KafkaSource<String> source = KafkaSource.<String>builder()
                    .setBootstrapServers(bootstrapServers)
                    .setTopics(financialDataTopic)
                    .setGroupId(groupId)
                    .setStartingOffsets(OffsetsInitializer.latest())
                    .setValueOnlyDeserializer(new SimpleStringSchema())
                    .build();

            // 创建数据流
            DataStream<String> stream = env.fromSource(
                    source,
                    WatermarkStrategy.noWatermarks(),
                    "Kafka Source");

            // 解析JSON数据并转换为FinancialIndicator对象
            DataStream<FinancialIndicator> financialStream = stream
                    .map(new FinancialIndicatorMapper())
                    .filter(indicator -> indicator != null);

            // 按用户ID分组，计算财务指标
            financialStream
                    .keyBy(FinancialIndicator::getUserId)
                    .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
                    .process(new FinancialIndicatorProcessor());

            // 启动Flink任务
            env.execute("Financial Indicator Job");
        } catch (Exception e) {
            log.error("启动财务指标Flink任务失败", e);
        }
    }
}